package com.wudl.sse;


import com.wudl.avro.SseAvro;
import com.wudl.utils.DateTimeUtil;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.flume.Context;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.io.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;



/**
 * @author ：wudl
 * @date ：Created in 2021-12-18 0:33
 * @description：
 * @modified By：
 * @version: 1.0
 */

public class SseMySource extends AbstractSource implements PollableSource, Configurable {
    private String host;
    private Integer port;
    private String userName;
    private String password;
    private String directory;
    private String fileName;
    private String localDirectory;
    private Integer delay;
    private Integer corePoolSize;
    private Integer maxPoolSize;
    private Long keepAliveTime;
    private Integer capacity;
    ThreadPoolExecutor threadPoolExecutor;

    @Override
    public Status process() throws EventDeliveryException {
        //3.判断是否是交易时间段
        long time = System.currentTimeMillis();
        if (time > DateTimeUtil.openTime && time < DateTimeUtil.closeTime) {
            //4.异步处理
            threadPoolExecutor.execute(new AsyncFlume());

            //5.设置延时时间
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return Status.READY;
    }
    /**
     * 创建线程
     */
    private class AsyncFlume implements Runnable {

        @Override
        public void run() {
            //下载ftp行情文件
            boolean bool = download(directory, localDirectory);
            try {
                if (bool == true) {
                    //1.读取文件获取行数据
                    InputStreamReader inputStreamReader = new InputStreamReader(new FileInputStream(localDirectory + "/" + fileName));
                    BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
                    String line;
                    int i = 0;
                    while ((line = bufferedReader.readLine()) != null) {
//2.获取首行，判断市场行情状态
                        String[] arr = line.split("\\|");
                        if (i == 0) {
                            //市场行情状态
                            String status = arr[8];
                            if (status.startsWith("E")) {
                                break;
                            }
                        } else {
                            //3.数据转换成avro对象
                            SseAvro sseAvro = transform(arr);
                            //4.序列化操作
                            byte[] serialize = serialize(sseAvro );
                            //5.发送管道
                            getChannelProcessor().processEvent(EventBuilder.withBody(serialize));
                        }
                        i++;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    private byte[] serialize(SseAvro sseAvro) {

        DatumWriter<SseAvro> specificDatumWriter = new SpecificDatumWriter<>(SseAvro.getClassSchema());
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();

        BinaryEncoder binaryEncoder = EncoderFactory.get().directBinaryEncoder(byteArrayOutputStream, null);
        try {
            specificDatumWriter.write(sseAvro, binaryEncoder);
        } catch (IOException e) {
            e.printStackTrace();
        }

        return byteArrayOutputStream.toByteArray();
    }
    private SseAvro transform(String[] arr) {
        SseAvro sseAvro = new SseAvro();
        sseAvro.setMdStreamID(arr[0].trim());
        sseAvro.setSecurityID(arr[1].trim());
        sseAvro.setSymbol(arr[2].trim());
        sseAvro.setTradeVolume(Long.valueOf(arr[3].trim()));
        sseAvro.setTotalValueTraded(Long.valueOf(arr[4].trim()));
        sseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
        sseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
        sseAvro.setHighPrice(Double.valueOf(arr[7].trim()));
        sseAvro.setLowPrice(Double.valueOf(arr[8].trim()));
        sseAvro.setTradePrice(Double.valueOf(arr[9].trim()));
        sseAvro.setClosePx(Double.valueOf(arr[10].trim()));
        sseAvro.setTradingPhaseCode("T11");
        sseAvro.setTimestamp(System.currentTimeMillis());
        return sseAvro;
    }
    private boolean download(String directory, String localDirectory) {

        boolean flag = false;
        //1.初始化ftp连接
        FTPClient ftpClient = initFtpClient();
        try {
            //2.切换工作目录
            ftpClient.changeWorkingDirectory(directory);
            //被动模式，服务端开放端口，用于数据传输
            ftpClient.enterLocalPassiveMode();
            //禁用服务端参与的验证，如果不禁用服务端会获取主机IP与提交的host进行匹配，不一致时会报错
            ftpClient.setRemoteVerificationEnabled(false);
            //3.获取工作目录的文件信息列表
            FTPFile[] ftpFiles = ftpClient.listFiles();
            for (FTPFile ftpFile : ftpFiles) {
                File file = new File(localDirectory + "/" + ftpFile.getName());
                FileOutputStream fos = new FileOutputStream(file);
                //4.文件输出
                ftpClient.retrieveFile(ftpFile.getName(), fos);
                fos.close();
            }
            ftpClient.logout();
            System.out.println("文件下载成功。。。。");
            //5.返回成功状态
            flag = true;
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (ftpClient.isConnected()) {
                try {
                    ftpClient.disconnect();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return flag;
    }

    /**
     * 1.初始化ftp连接
     */
    private FTPClient initFtpClient() {
        FTPClient ftpClient = new FTPClient();
        ftpClient.setControlEncoding("UTF-8");
        try {
            ftpClient.connect(host, port);
            ftpClient.login(userName, password);
            //FTP服务器检查连接是否成功,220表示连接成功，230登陆成功，221再见
            int replyCode = ftpClient.getReplyCode();
            //判断连接是否成功响应
            if (!FTPReply.isPositiveCompletion(replyCode)) {
                System.out.println("connect ftp sever failed ");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return ftpClient;
    }
    @Override
    public void configure(Context context) {
        host = context.getString("host");
        port = context.getInteger("port");
        userName = context.getString("userName");
        password = context.getString("password");
        directory = context.getString("ftpDirectory");
        fileName = context.getString("fileName");
        localDirectory = context.getString("localDirectory");
        delay = context.getInteger("delay");
        corePoolSize = context.getInteger("corePoolSize");
        maxPoolSize = context.getInteger("maxPoolSize");
        keepAliveTime = context.getLong("keepAliveTime");
        capacity = context.getInteger("capacity");

        /*
         *初始化线程池
         */
        threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(capacity));

    }

}
